package com.wfm.files;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 通过多线程处理大文件，实现多线程并行处理，提高处理效率
 * 主线程创建多个子线程，每个子线程处理一个任务块，然后等待所有子线程完成，再合并结果。
 * 主线程负责计算每行的起始位置和长度，然后分配给线程池里的线程
 * 线程池里的线程分别处理不同区域的文件，所以前提要知道最终写入文件的大小。比如源文件123，修改成1,2,3那么从原来的3变成了5，这里没有考虑\r\n。
 */
public class FileProcessor {
    static CountDownLatch latch;

    static class LineInfo {
        long start;
        int length;

        LineInfo(long start, int length) {
            this.start = start;
            this.length = length;
        }
    }

    static class TaskBlock {
        int startLine;
        int endLine;

        TaskBlock(int startLine, int endLine) {
            this.startLine = startLine;
            this.endLine = endLine;
        }
    }

    public static void process(Path inputPath, Path outputPath, int linesPerBlock, int numThreads) throws IOException, InterruptedException {
        // 读取输入文件，记录每行的起始位置和长度
        List<LineInfo> lines = readLineInfos(inputPath);

        // 计算每行输出的长度和偏移量
        long[] outputOffsets = new long[lines.size()];
        int[] outputLengths = new int[lines.size()];
        calculateOutputMetadata(lines, outputOffsets, outputLengths);

        // 创建输出文件并设置大小
        createOutputFile(outputPath, outputOffsets, outputLengths);

        // 分块任务
        List<TaskBlock> taskBlocks = createTaskBlocks(lines.size(), linesPerBlock);

        latch = new CountDownLatch(taskBlocks.size());
        // 使用线程池处理任务块
        ExecutorService executor = Executors.newFixedThreadPool(numThreads);
        for (TaskBlock block : taskBlocks) {
            executor.submit(() -> processBlock(block, inputPath, outputPath, lines, outputOffsets, outputLengths));
        }
        latch.await();
        executor.shutdown();
        if (!executor.awaitTermination(1, TimeUnit.HOURS)) {
            executor.shutdownNow();
        }
    }

    private static List<FileProcessor.LineInfo> readLineInfos(Path inputPath) throws IOException {
        List<FileProcessor.LineInfo> lines = new ArrayList<>();
        try (FileChannel channel = FileChannel.open(inputPath, StandardOpenOption.READ)) {
            ByteBuffer buffer = ByteBuffer.allocate(8192);
            long filePosition = 0;
            int lineStartInBuffer = 0;
            boolean hasRemaining = false;

            while (channel.read(buffer) != -1 || hasRemaining) {
                buffer.flip();

                // 合并上次剩余数据和新数据
                byte[] combined = new byte[buffer.remaining()];
                buffer.get(combined);
                buffer.clear();
                buffer.put(combined);
                buffer.flip();

                for (int i = 0; i < buffer.limit(); i++) {
                    if (buffer.get(i) == '\n') {
                        // 计算绝对起始位置
                        long lineStart = filePosition + lineStartInBuffer;
                        int lineLength = i - lineStartInBuffer + 1;

                        lines.add(new FileProcessor.LineInfo(lineStart, lineLength));

                        // 更新下一行的起始位置
                        lineStartInBuffer = i + 1;
                    }
                }

                // 处理剩余未完成的行
                if (buffer.limit() > 0) {
                    hasRemaining = buffer.get(buffer.limit() - 1) != '\n';
                    filePosition += lineStartInBuffer;

                    // 保存未处理的数据
                    byte[] remaining = new byte[buffer.limit() - lineStartInBuffer];
                    buffer.position(lineStartInBuffer);
                    buffer.get(remaining);

                    buffer.clear();
                    buffer.put(remaining);
                    lineStartInBuffer = 0;
                } else {
                    buffer.clear();
                }
            }

            // 处理最后一行没有换行符的情况
            if (lineStartInBuffer > 0) {
                lines.add(new FileProcessor.LineInfo(filePosition, buffer.limit() - lineStartInBuffer));
            }
        }
        return lines;
    }

    private static void calculateOutputMetadata(List<LineInfo> lines, long[] outputOffsets, int[] outputLengths) {
        long currentOffset = 0;
        for (int i = 0; i < lines.size(); i++) {
            int inputLength = lines.get(i).length;
            outputLengths[i] = (inputLength - 1) * 2;//如果考虑\r\n，需要－1
            outputOffsets[i] = currentOffset;
            currentOffset += outputLengths[i];
        }
    }

    private static void createOutputFile(Path outputPath, long[] outputOffsets, int[] outputLengths) throws IOException {
        if (outputOffsets.length == 0) return;
        long totalSize = outputOffsets[outputOffsets.length - 1] + outputLengths[outputLengths.length - 1];
        try (FileChannel channel = FileChannel.open(outputPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
            channel.truncate(totalSize);
        }
    }

    private static List<TaskBlock> createTaskBlocks(int totalLines, int linesPerBlock) {
        List<TaskBlock> blocks = new ArrayList<>();
        for (int start = 0; start < totalLines; start += linesPerBlock) {
            int end = Math.min(start + linesPerBlock - 1, totalLines - 1);
            blocks.add(new TaskBlock(start, end));
        }
        return blocks;
    }

    private static void processBlock(TaskBlock block, Path inputPath, Path outputPath,
                                     List<LineInfo> lines, long[] outputOffsets, int[] outputLengths) {
        int startLine = block.startLine;
        int endLine = block.endLine;

        long inputStart = lines.get(startLine).start;
        long inputEnd = lines.get(endLine).start + lines.get(endLine).length;
        long inputSize = inputEnd - inputStart;

        long outputStartOffset = outputOffsets[startLine];
        long outputTotalLength = outputOffsets[endLine] + outputLengths[endLine] - outputStartOffset;

        System.out.println(Thread.currentThread().getName() +" Processing block: " + startLine + " to " + endLine);
        System.out.println(Thread.currentThread().getName() +" Input start: " + inputStart + ", Input end: " + inputEnd+ ", Input size: " + inputSize);
        System.out.println(Thread.currentThread().getName() +" Output start offset: " + outputStartOffset + ", Output total length: " + outputTotalLength);

        try (FileChannel inputChannel = FileChannel.open(inputPath, StandardOpenOption.READ);
             FileChannel outputChannel = FileChannel.open(outputPath, StandardOpenOption.CREATE,StandardOpenOption.WRITE,StandardOpenOption.READ)) {

            MappedByteBuffer inputBuffer = inputChannel.map(FileChannel.MapMode.READ_ONLY, inputStart, inputSize);
            MappedByteBuffer outputBuffer = outputChannel.map(FileChannel.MapMode.READ_WRITE, outputStartOffset, outputTotalLength);

            int currentOutputPos = 0;
            for (int i = startLine; i <= endLine; i++) {
                LineInfo line = lines.get(i);
                int lineRelativePos = (int) (line.start - inputStart);
                byte[] lineBytes = new byte[line.length];
                inputBuffer.position(lineRelativePos);
                inputBuffer.get(lineBytes);

                // 处理行内容
                String lineStr = new String(lineBytes, 0, lineBytes.length, StandardCharsets.UTF_8);
                StringBuilder sb = new StringBuilder();
                for (int j = 0; j < lineStr.length()-1; j++) {//如果考虑\r\n，需要再－1变成-2
                    if (j > 0) {
                        sb.append(',');
                    }
                    sb.append(lineStr.charAt(j));
                }
//                sb.append('\r');//如果考虑\r\n，需要去掉
                sb.append('\n');
                byte[] processedBytes = sb.toString().getBytes(StandardCharsets.UTF_8);

                if (processedBytes.length != outputLengths[i]) {
                    throw new RuntimeException("Processed length mismatch for line " + i);
                }

                outputBuffer.position(currentOutputPos);
                outputBuffer.put(processedBytes);
                currentOutputPos += processedBytes.length;
            }
            outputBuffer.force();
            latch.countDown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            Path inputPath = Path.of("input.txt");
            Path outputPath = Path.of("output.txt");
            int linesPerBlock = 2;
            int numThreads = 5;
            process(inputPath, outputPath, linesPerBlock, numThreads);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}